package edu.purdue.cs505;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

public class ReliableBroadcastImpl implements ReliableBroadcast,
		ReliableChannelReceiver {
	private Process myProcess;
	private Collection<Process> otherProcesses;
	private BroadcastReceiver callback;
	private Set<Message> receivedMessages;

	public ReliableBroadcastImpl() {
		myProcess = null;
		callback = null;
		otherProcesses = new HashSet<Process>();
		receivedMessages = new HashSet<Message>();
	}

	@Override
	public void init(Process currentProcess) {
		myProcess = currentProcess;
	}

	@Override
	public void addProcess(Process p) {
	  /*
	   * The modifier is used to generate a unique pair of local/remote ports for the channels
	   * between each process. We can't simply reuse the same port for each communication since
	   * if there are more than 2 nodes several will end up conflicting
	   */
	  int modifier = myProcess.getIndex() + p.getIndex();
		ReliableChannel newChannel = new ReliableChannelImpl(myProcess.getIP(),
				myProcess.getPort() + modifier);
		newChannel.setMyProcessId(myProcess.getProcessID());
		newChannel.init(p.getIP(), p.getPort() + modifier);
		newChannel.rlisten(this);
		p.setChannel(newChannel);
		otherProcesses.add(p);
	}

	@Override
	public void rbroadcast(Message m) {
		m.setProcessID(myProcess.getProcessID());
		deliverAndBroadcast(m);
	}

	@Override
	public void rblisten(BroadcastReceiver m) {
		callback = m;
	}

	/*
	 * Used to define our own callback when we receive a message from the
	 * ReliableChannel
	 */
	@Override
	public synchronized void rreceive(Message m) {
		if (!receivedMessages.contains(m)) {
			deliverAndBroadcast(m);
		}
	}

	@Override
	public void halt() {
		for (Process p : otherProcesses) {
			p.getChannel().halt();
		}
	}

	private synchronized void deliverAndBroadcast(Message m) {
		/*
		 * Simple algo: - p sends to every other process including itself - every
		 * process which receives m for the first time sends it to every other
		 * process except the sender and delivers it
		 */

		// first send to myself and deliver
		receivedMessages.add(m);
		callback.rdeliver(m);

		// send to all others
		for (Process p : otherProcesses) {
			// if the process is the sender, don't send to it
			if (!m.getProcessID().equals(p.getProcessID())) {
				ReliableChannel rc = p.getChannel();
				rc.rsend(m);
			}
		}
	}
}
